#include "thread.h"
#include "memalloc.h"
#include "common.h"
#include "threadtask.h"

#ifdef _WIN32
#include <process.h>

namespace Cross
{

Thread::Thread()
:thread_is_running(false)
,mThread(NULL)
{

}
void Thread::start()
{
  if(NULL==mThread)
  {
    mThread=(HANDLE)_beginthreadex(NULL,0,thread_func,this,0,NULL);
    if(NULL==mThread)
      throw String("_beginthreadex,thread_func,fail!");
    thread_is_running=true;
  }
}
void Thread::stop()
{
  if(NULL!=mThread)
  {
    thread_is_running=false;
    WaitForSingleObject(mThread,INFINITE);
    CloseHandle(mThread);
    mThread=NULL;
  }
}
unsigned int __stdcall Thread::thread_func(void *p)
{
  ((Thread *)p)->run();
  _endthreadex(0);
  return 0;
}

struct ThreadParameter
{
    void (*func)(void*);
    void* p;
};

static unsigned int __stdcall ThreadFunc(void* p)
{
    ThreadParameter* param=(ThreadParameter*)p;
    void (*func)(void*);
    func=param->func;
    p=param->p;
    delete param;
    func(p);
    _endthreadex(0);
    return 0;
}

void StartThread(void (*func)(void*),void*p)
{
    ThreadParameter* param=new ThreadParameter;
    param->func=func;
    param->p=p;
    HANDLE hdl=(HANDLE)_beginthreadex(NULL,0,ThreadFunc,param,0,NULL);
    CloseHandle(hdl);
}

void StartThread(unsigned int (__stdcall *func)(void*),void* p)
{
    HANDLE hdl=(HANDLE)_beginthreadex(NULL,0,func,p,0,NULL);
    CloseHandle(hdl);
}

}

#else//_WIN32

namespace Cross
{

Thread::Thread()
:thread_is_running(false)
{

}
void Thread::start()
{
  if(!thread_is_running)
  {  
    pthread_create(&mthreadid,NULL,thread_func,this);
    thread_is_running=true;
  }
}
void Thread::stop()
{
  if(thread_is_running)
  {
    thread_is_running=false;
    pthread_join(mthreadid,NULL);
  }
}
void *Thread::thread_func(void *p)
{
  ((Thread *)p)->run();
  pthread_exit(0);
  return NULL;
}

struct ThreadParameter
{
    void (*func)(void*);
    void* p;
};

static void* ThreadFunc(void* p)
{
    ThreadParameter* param=(ThreadParameter*)p;
    void (*func)(void*);
    func=param->func;
    p=param->p;
    delete param;
    func(p);
    pthread_exit(0);
    return NULL;
}

void StartThread(void (*func)(void*),void*p)
{
    pthread_t mthreadid;
    ThreadParameter* param=new ThreadParameter;
    param->func=func;
    param->p=p;
    pthread_create(&mthreadid,NULL,ThreadFunc,(void*)param);
}

}

#endif//_WIN32

namespace Cross
{
inline void TaskProcessor::ReturnBack(ITask* task)
{
    Locker lock(&_retMutex);
    list_add_tail(&task->_lst,&_retBack);
}

class TaskThread:public Thread
{
public:
    TaskThread(TaskProcessor* taskProcessor)
    {
        INIT_LIST_HEAD(&_taskQueue);
        _taskProcessor=taskProcessor;
    }
    inline void AcceptTask(ITask* task)
    {
        Locker lock(&_taskMutex);
        list_add_tail(&task->_lst,&_taskQueue);
    }
protected:
    virtual void run()
    {
        while(thread_is_running)
        {
            LIST_HEAD(tmpqueue);
            {
                Locker lock(&_taskMutex);
                list_splice_init(&_taskQueue,&tmpqueue);
            }
            list_head* iter;
            list_head* next;
            list_for_each_safe(iter,next,&tmpqueue)
            {
                ITask* task=list_entry(iter,ITask,_lst);
                if(task->DoThread())
                {
                    delete task;
                }
                else
                {
                    _taskProcessor->ReturnBack(task);
                }
            }
        }
    }
private:
    list_head _taskQueue;
    UserMutex _taskMutex;
    TaskProcessor* _taskProcessor;
};

TaskProcessor::~TaskProcessor()
{
    list_head* iter;
    list_head* next;
    list_for_each_safe(iter,next,&_retBack)
    {
        ITask* task=list_entry(iter,ITask,_lst);
        delete task;
    }
    INIT_LIST_HEAD(&_retBack);
    for(int i=0;i<_thNum;++i)
    {
        _thArray[i].~TaskThread();
    }
    GetAllocator()->free(_thArray,sizeof(TaskThread)*_thNum);
    _thArray=NULL;
}

TaskProcessor::TaskProcessor(int threadNum)
{
    _thNum=threadNum;
    _thArray=(TaskThread*)GetAllocator()->alloc(sizeof(TaskThread)*threadNum);
    for(int i=0;i<threadNum;++i)
    {
        new(_thArray+i) TaskThread(this);
    }
    INIT_LIST_HEAD(&_retBack);
}

void TaskProcessor::Start()
{
    for(int i=0;i<_thNum;++i)
    {
        _thArray[i].start();
    }
}

void TaskProcessor::Stop()
{
    for(int i=0;i<_thNum;++i)
    {
        _thArray[i].stop();
    }
}

void TaskProcessor::PostTask(ITask* task)
{
    int index=GetRandom(0,_thNum);
    assert(index>=0&&index<_thNum);
    _thArray[index].AcceptTask(task);
}

void TaskProcessor::Tick()
{
    LIST_HEAD(tmpqueue);
    {
        Locker lock(&_retMutex);
        list_splice_init(&_retBack,&tmpqueue);
    }
    list_head* iter;
    list_head* next;
    list_for_each_safe(iter,next,&tmpqueue)
    {
        ITask* task=list_entry(iter,ITask,_lst);
        task->DoReturn();
        delete task;
    }
}

}